package com.xiaour.spring.boot.rocketmq.consumer;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.expression.ParseException;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;

@Component
public class LogisticsConsumer {
	  /**
     * 消费者
     */
    @Value("${apache.rocketmq.consumer.pushConsumer}")
    private String pushConsumer;

    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
    
    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.consumer.LogisticsConsumer}")
    private String consumerGroup;
    
    private DefaultMQPullConsumer consumer;
    
    @PostConstruct
    public void defaultMQProducer() {
    	
    	
        consumer = new DefaultMQPullConsumer(consumerGroup);
        
        consumer.setNamesrvAddr(namesrvAddr);
        

        try {
        	consumer.start();
            System.out.println("-------->:friendConsumer启动了");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    public String viewMessage(String OffsetMsgId) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
    	 MessageExt msg = consumer.viewMessage(OffsetMsgId);
        return "{\"OffsetMsgId\":\""+OffsetMsgId+"\", \"msg\":\""+new String(msg.getBody())+"\"}";
    }
    
    
    public String pull(String topic, String keys, String begin, String end) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        
    	
    	try {
        	
        	
    		JSONObject resultJson = new JSONObject();
        	
            QueryResult queryresult = consumer.queryMessage("LogisticsInfor", keys, 20, dateToStamp(begin), dateToStamp(end));
            
            String lastupdate = String.valueOf(queryresult.getIndexLastUpdateTimestamp());
            
            System.out.printf("%s%n", stampToDate(lastupdate));
            System.out.printf("%s%n", queryresult.getMessageList());
            
           
            List<MessageExt> messageExtList = queryresult.getMessageList();
            
            int i =1;
            for (MessageExt m : messageExtList) {
            	//String msg = new String(m.getBody());
                //System.out.println(msg);
                resultJson.put(String.valueOf(i), new String(m.getBody()));
                i++;
            }
            System.out.println(resultJson);
            resultJson.put("num", messageExtList.size());
            resultJson.put("lastupdate", stampToDate(lastupdate));
            resultJson.put("userid", keys);
            resultJson.put("topic", "LogisticsInfor");
            
            return resultJson.toString();
        } catch (Exception e) {
            e.printStackTrace();
        }
		return "ERROR";
        
      
        
        
   }
    /* 
     * 将时间转换为时间戳
     */    
    public static long dateToStamp(String s) throws ParseException, java.text.ParseException{
        //String res;
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = simpleDateFormat.parse(s);
        long ts = date.getTime();
        //res = String.valueOf(ts);
        //return res;
        return ts;
    }

    /* 
    * 将时间戳转换为时间
    */
   public static String stampToDate(String s){
       String res;
       SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
       long lt = new Long(s);
       Date date = new Date(lt);
       res = simpleDateFormat.format(date);
       return res;
   }


}
